package com.cmbyc.Sink

import com.cmbyc.JDBC.sink.Sink1
import com.cmbyc.JDBC.source.Student
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

/**
 *
 * @program: com.cmbyc.Sink
 * @author: YCLW058
 * @create: 2021-05-25 11:19
 * @decsription:
 *
 * */

object MYSQLSink {
  def main(args: Array[String]): Unit = {
    //1 env
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val tenv = StreamTableEnvironment.create(env)
    //2 source
    val inputpath="data/input/test"
    val inputStream=env.readTextFile(inputpath)
    //3 transforamtion
    // 转化为样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data.split(",")
        Student(arr(0).toInt, arr(1), arr(2).toInt)
      })

    //4 sink
    dataStream.print()
    dataStream.addSink(new Sink1)
    //5 execute
    env.execute()
  }


}
